package avicit.bdp.dcs.json.service;

import avicit.bdp.common.utils.DateUtils;
import avicit.bdp.common.utils.enums.JobType;
import avicit.bdp.common.utils.redis.RedisCacheHelper;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.constant.ProcessConstants;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.task.service.TaskService;
import avicit.bdp.dcs.tools.ds.NodeConnect;
import avicit.bdp.dcs.tools.enums.Enable;
import avicit.bdp.dcs.tools.enums.SyncCondition;
import avicit.bdp.dcs.tools.enums.SyncMode;
import avicit.bdp.dcs.utils.DcsConstants;
import avicit.bdp.dcs.utils.DcsErrorCode;
import avicit.bdp.dds.dispatch.enums.*;
import avicit.bdp.dds.dispatch.model.ProcessData;
import avicit.bdp.dds.dispatch.model.TaskNode;
import avicit.bdp.dds.dispatch.process.Property;
import avicit.bdp.dds.dispatch.task.conditions.ConditionsParameters;
import avicit.bdp.dds.dispatch.task.datax.DataxParameters;
import avicit.bdp.dds.dispatch.task.http.HttpParameters;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.*;

/**
 * @author LIYB json构建实现类
 */
@Service
@Transactional
public class StructureTaskJsonService extends TaskJsonService{

  private static final Logger logger = LoggerFactory.getLogger(StructureTaskJsonService.class);

  @Resource private TaskService taskService;
  @Resource private RedisCacheHelper redisCacheHelper;

  /**
   * 构建Dolphinscheduler流程定义json数据和流程定义节点连接信息json 1、jobId等于Dolphinscheduler流程定义ID
   * 2、增量同步：1，全量同步：2，数据迁移：3 3、在迁移和同步过程中有创建表的情况，需要定义HTTP类型的前置节点，该节点用于回调数据采集系统中创建表服务
   * 4、有多节点情况，需要定义节点的连接信息，即构建t_ds_process_definition表connects字段数据 5、数据采集系统中job等同于DS中流程定义
   *
   * @param jobDTO
   * @return
   * @throws Exception
   */
  public Map<String, String> buildDsJson(JobDTO jobDTO) throws Exception {
    Map<String, String> map = new HashMap<>(4);

    // 查询job下的任务列表
    TaskDTO taskDTOQuery = new TaskDTO();
    taskDTOQuery.setJobId(jobDTO.getId());
    taskDTOQuery.setFlag(Enable.ZERO.getCode());
    List<TaskDTO> taskList = taskService.selectList(taskDTOQuery);

    if (CollectionUtils.isNotEmpty(taskList)) {
      ProcessData processData = new ProcessData();
      List<TaskNode> taskNodeList = new LinkedList<>();
      List<NodeConnect> nodeConnectList = new LinkedList<>();

      List<List<TaskDTO>> blocks = splitIntoBlocks(taskList);

      for(List<TaskDTO> list : blocks){
        TaskNode taskNode = null;
        for (TaskDTO task : list) {
          // 数据迁移只选择“表结构迁移”，则不创建DATAX任务节点
          if (JobType.TRANSFER.getCode() == jobDTO.getJobType()
                  && Enable.ONE.getDescp().equals(jobDTO.getTransferType())) {
            // 仅创建表的节点
            createTableNode(task, taskNodeList);
          } else {
            taskNode = createDataxNode(jobDTO, task, taskNodeList, nodeConnectList, taskNode);
          }
        }
      }

      processData.setTasks(taskNodeList);
      map.put(ProcessConstants.PROCESS_DEFINITION_JSON, JSON.toJSONString(processData));
      map.put(ProcessConstants.CONNECTS, JSON.toJSONString(nodeConnectList));
    }
    return map;
  }

  /** 创建Datax节点 */
  private TaskNode createDataxNode(
          JobDTO jobDTO, TaskDTO task, List<TaskNode> taskNodeList, List<NodeConnect> nodeConnectList, TaskNode lastTaskNode)
          throws Exception {
    Random random = new Random();
    List<String> preTaskList = new ArrayList<>();

    // 创建前置处理节点
    String preProcessNodeId =
            ProcessConstants.TASKS + "-preProcess" + Constants.SUBTRACT_STRING + random.nextInt(MAX_RANDOM);
    TaskNode preProcessNode = new TaskNode();
    preProcessNode.setId(preProcessNodeId);
    preProcessNode.setType(TaskType.HTTP.name());
    preProcessNode.setName(task.getId() + Constants.SUBTRACT_STRING + PRE_PROCESS_NODE);
    preProcessNode.setDesc(Constants.STRING_BLANK);
    preProcessNode.setRunFlag(ProcessConstants.NORMAL);
    preProcessNode.setMaxRetryTimes(0);
    preProcessNode.setRetryInterval(1);
    preProcessNode.setTaskInstancePriority(Priority.MEDIUM);
    preProcessNode.setWorkerGroup(ProcessConstants.DEFAULT);
    preProcessNode.setConditionResult(JSON.toJSONString(new ConditionsParameters()));

    HttpParameters httpParameters = new HttpParameters();
    httpParameters.setUrl(preProcessUrl + Constants.SINGLE_SLASH + task.getId());
    httpParameters.setHttpMethod(HttpMethod.GET);
    preProcessNode.setParams(JSON.toJSONString(httpParameters));
    taskNodeList.add(preProcessNode);

    // 任务串行执行
    if(lastTaskNode != null){
      NodeConnect preProcessNodeConnect = new NodeConnect();
      preProcessNodeConnect.setEndPointSourceId(lastTaskNode.getId());
      preProcessNodeConnect.setEndPointTargetId(preProcessNodeId);
      nodeConnectList.add(preProcessNodeConnect);

      preTaskList = new ArrayList<>();
      preTaskList.add(lastTaskNode.getName());
      preProcessNode.setPreTasks(JSON.toJSONString(preTaskList));
    }

    // 创建dataX采集数据节点
    String taskNodeId =
            ProcessConstants.TASKS + "-dataX" + Constants.SUBTRACT_STRING + random.nextInt(MAX_RANDOM);
    TaskNode taskNode = new TaskNode();
    taskNode.setId(taskNodeId);

    if (JobType.SYNC.getCode() == jobDTO.getJobType()
            && SyncMode.COMMON_INCREMENT_SYNC.getCode() == task.getSyncMode()) {
      buildDataxIncrementNode(taskNode, jobDTO, task);
    } else {
      buildDataxNode(taskNode, jobDTO, task);
    }

    // 需要创建表结构，则创建HTTP类型的前置节点，该节点用于回调数据采集系统中创建表服务
    // 获取job是否需创建表
    boolean createTable = checkCreateTable(jobDTO);
    if (createTable) {
      String createTableNodeId =
              ProcessConstants.TASKS + "-createTable" + Constants.SUBTRACT_STRING + random.nextInt(MAX_RANDOM);
      TaskNode createTableNode = new TaskNode();
      createTableNode.setId(createTableNodeId);
      createTableNode.setType(TaskType.HTTP.name());
      createTableNode.setName(task.getId() + Constants.SUBTRACT_STRING + task.getSrcTableName());
      createTableNode.setDesc(Constants.STRING_BLANK);
      createTableNode.setRunFlag(ProcessConstants.NORMAL);
      createTableNode.setMaxRetryTimes(0);
      createTableNode.setRetryInterval(1);
      createTableNode.setTaskInstancePriority(Priority.MEDIUM);
      createTableNode.setWorkerGroup(ProcessConstants.DEFAULT);

      createTableNode.setConditionResult(JSON.toJSONString(new ConditionsParameters()));

      httpParameters = new HttpParameters();
      httpParameters.setUrl(createTableUrl + Constants.SINGLE_SLASH + task.getId());
      httpParameters.setHttpMethod(HttpMethod.GET);
      createTableNode.setParams(JSON.toJSONString(httpParameters));
      taskNodeList.add(createTableNode);

      // 构建Dolphinscheduler流程定义节点连接信息json,[{"endPointSourceId":"tasks-44147","endPointTargetId":"tasks-70818"}]
      NodeConnect nodeConnect = new NodeConnect();
      nodeConnect.setEndPointSourceId(preProcessNodeId);
      nodeConnect.setEndPointTargetId(createTableNodeId);
      nodeConnectList.add(nodeConnect);

      preTaskList = new ArrayList<>();
      preTaskList.add(preProcessNode.getName());
      createTableNode.setPreTasks(JSON.toJSONString(preTaskList));

      nodeConnect = new NodeConnect();
      nodeConnect.setEndPointSourceId(createTableNodeId);
      nodeConnect.setEndPointTargetId(taskNodeId);
      nodeConnectList.add(nodeConnect);

      preTaskList = new ArrayList<>();
      preTaskList.add(createTableNode.getName());
      taskNode.setPreTasks(JSON.toJSONString(preTaskList));
    } else {
      NodeConnect nodeConnect = new NodeConnect();
      nodeConnect.setEndPointSourceId(preProcessNodeId);
      nodeConnect.setEndPointTargetId(taskNodeId);
      nodeConnectList.add(nodeConnect);

      preTaskList = new ArrayList<>();
      preTaskList.add(preProcessNode.getName());
      taskNode.setPreTasks(JSON.toJSONString(preTaskList));
    }
    taskNodeList.add(taskNode);
    return taskNode;
  }

  /** 获取job是否需创建表 */
  private boolean checkCreateTable(JobDTO jobDTO) {
    boolean createTable = false;
    if (JobType.TRANSFER.getCode() == jobDTO.getJobType()) {
      if (jobDTO.getTransferType().contains(Enable.ONE.getDescp())) {
        createTable = true;
      }
    } else if (JobType.SYNC.getCode() == jobDTO.getJobType()) {
      createTable = true;
    } else {
      logger.warn(DcsErrorCode.UNSUPPORTED_JOB_TYPE.getDesc() + jobDTO.getJobType());
    }

    return createTable;
  }

  /**
   * 构造DATAX采集节点
   *
   * @param taskNode
   * @param jobDTO
   * @param task
   */
  private void buildDataxNode(TaskNode taskNode, JobDTO jobDTO, TaskDTO task) {
    taskNode.setType(TaskType.DATAX.name());
    taskNode.setName(task.getId());
    taskNode.setDesc(Constants.STRING_BLANK);
    taskNode.setRunFlag(ProcessConstants.NORMAL);
    taskNode.setMaxRetryTimes(0);
    taskNode.setRetryInterval(1);
    taskNode.setTaskInstancePriority(Priority.MEDIUM);
    taskNode.setWorkerGroup(ProcessConstants.DEFAULT);

    ConditionsParameters conditionsParameters = new ConditionsParameters();
    taskNode.setConditionResult(JSON.toJSONString(conditionsParameters));

    DataxParameters dataxParameters = new DataxParameters();
    dataxParameters.setCustomConfig(1);
    dataxParameters.setJson(task.getDataxJson());
    dataxParameters.setDataSource(jobDTO.getSrcDatasourceId());
    dataxParameters.setSourceTable(task.getSrcTableName());
    dataxParameters.setDataTarget(jobDTO.getDstDatasourceId());
    dataxParameters.setTargetTable(task.getDstTableName());
    dataxParameters.setColumnSource(getSourceColumn(task));
    dataxParameters.setColumnTarget(getDstColumn(task));

    // datax增量同步方式（基于时间戳字段）
    if (JobType.SYNC.getCode() == jobDTO.getJobType()
            && SyncMode.DATAX_INCREMENT_SYNC.getCode() == task.getSyncMode()) {
      // 设置源表增量字段
      dataxParameters.setSourceIncrementColumn(task.getFieldName());
      // 如果是时间字段更新的增量同步方式
      if (task.getSyncConditionField() != null
              && task.getSyncConditionField() == SyncCondition.TIME.getCode()) {
        dataxParameters.setIncrementMode(IncrementModeEnum.INCREMENT_TIME_FIELD);
        Date syncTimePoint = task.getSyncTimePoint();
        // 用户自定义数据同步时间，如果存在，保存时间到redis或其他地方，类似checkpoint机制，作为上次同步时间记录
        // todo 如果用户修改了syncTimePoint，已同步任务数据如何处理？
        if (syncTimePoint != null) {
          String cacheKey = Constants.REDIS_KEY_INCREMENT_LAST_SYNC_TIME + task.getId();
          String syncTimePointStr = DateUtils.getDateByString(syncTimePoint);
          redisCacheHelper.set(cacheKey, syncTimePointStr);
          logger.info("基于时间戳字段增量同步方式，设置用户同步开始时间 = {}, 缓存key = {}", syncTimePointStr, cacheKey);
        }
      }
    }

    int jobType = 1;
    if (JobType.TRANSFER.getCode() == jobDTO.getJobType()) {
      jobType = 1;
    } else if (JobType.SYNC.getCode() == jobDTO.getJobType()) {
      if (SyncMode.TOTAL_SYNC.getCode() == task.getSyncMode()) {
        jobType = 2;
      } else if (SyncMode.DATAX_INCREMENT_SYNC.getCode() == task.getSyncMode()) {
        jobType = 3;
      }
    } else {
      // 不支持
    }

    Property property = new Property();
    property.setProp(DcsConstants.JOB_TYPE);
    property.setDirect(Direct.IN);
    property.setType(DataType.VARCHAR);
    property.setValue(String.valueOf(jobType));
    List<Property> localParams = new ArrayList<>();
    localParams.add(property);

    dataxParameters.setLocalParams(localParams);
    taskNode.setParams(JSON.toJSONString(dataxParameters));
  }

  /** 构造DATAX增量采集节点 */
  private void buildDataxIncrementNode(TaskNode taskNode, JobDTO jobDTO, TaskDTO task) {
    taskNode.setType(TaskType.DATAXINCREMENT.name());
    taskNode.setName(task.getId());
    taskNode.setDesc(Constants.STRING_BLANK);
    taskNode.setRunFlag(ProcessConstants.NORMAL);
    taskNode.setMaxRetryTimes(0);
    taskNode.setRetryInterval(1);
    taskNode.setTaskInstancePriority(Priority.MEDIUM);
    taskNode.setWorkerGroup(ProcessConstants.DEFAULT);

    ConditionsParameters conditionsParameters = new ConditionsParameters();
    taskNode.setConditionResult(JSON.toJSONString(conditionsParameters));

    DataxParameters dataxParameters = new DataxParameters();
    dataxParameters.setCustomConfig(1);
    dataxParameters.setJson(task.getDataxJson());

    dataxParameters.setDataSource(jobDTO.getSrcDatasourceId());
    dataxParameters.setSourceTable(task.getSrcTableName());
    dataxParameters.setDataTarget(jobDTO.getDstDatasourceId());
    dataxParameters.setTargetTable(task.getDstTableName());
    dataxParameters.setColumnSource(getSourceColumn(task));
    dataxParameters.setColumnTarget(getDstColumn(task));

    // 唯一约束（主键）
    if (StringUtils.isNotEmpty(task.getFieldName())) {
      JSONObject json = JSONObject.parseObject(task.getFieldName());
      dataxParameters.setSourceKeyColumn(json.getString(Constants.UNIQUE_KEY));
      dataxParameters.setSourceIncrementColumn(json.getString(Constants.INCREMENTAL_COLUMN));
      dataxParameters.setTargetKeyColumn(json.getString(Constants.UNIQUE_KEY));
      dataxParameters.setTargetIncrementColumn(json.getString(Constants.INCREMENTAL_COLUMN));
    }

    taskNode.setParams(JSON.toJSONString(dataxParameters));
  }

  /** 仅创建表的节点 */
  private void createTableNode(TaskDTO task, List<TaskNode> taskNodeList) {
    Random random = new Random();
    String createTableNodeId =
            ProcessConstants.TASKS + Constants.SUBTRACT_STRING + random.nextInt(MAX_RANDOM);
    TaskNode createTableNode = new TaskNode();
    createTableNode.setId(createTableNodeId);
    createTableNode.setType(TaskType.HTTP.name());
    createTableNode.setName(task.getId() + Constants.SUBTRACT_STRING + task.getSrcTableName());
    createTableNode.setDesc(Constants.STRING_BLANK);
    createTableNode.setRunFlag(ProcessConstants.NORMAL);
    createTableNode.setMaxRetryTimes(0);
    createTableNode.setRetryInterval(1);
    createTableNode.setTaskInstancePriority(Priority.MEDIUM);
    createTableNode.setWorkerGroup(ProcessConstants.DEFAULT);

    createTableNode.setConditionResult(JSON.toJSONString(new ConditionsParameters()));

    HttpParameters httpParameters = new HttpParameters();
    httpParameters.setUrl(createTableUrl + Constants.SINGLE_SLASH + task.getId());
    httpParameters.setHttpMethod(HttpMethod.GET);
    createTableNode.setParams(JSON.toJSONString(httpParameters));
    taskNodeList.add(createTableNode);
  }

}
